/*
    异步工作器(双缓冲区)
    1、外界将任务添加到输入缓冲区
    2、异步线程对处理缓冲区中的数据处理，若处理缓冲区没有数据则交换缓冲区
*/
#ifndef __MY_LOPPER_H__
#define __MY_LOPPER_H__
#include "Buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <functional>
#include <atomic>
using namespace std;
namespace Lopper
{
    enum class AsyncType
    {
        ASYNC_SAFE,
        ASYNC_UNSAFE
    };
    class AsyncLopper
    {
        using Func = function<void(LogBuffer::Buffer &)>;

    public:
        using ptr = shared_ptr<AsyncLopper>;
        AsyncLopper(const Func &cb,
                    AsyncType type = AsyncType::ASYNC_SAFE)
            : _stop(false), _lopper_type(type),
              _thread(&AsyncLopper::threadFunc, this), Call_Back(cb) {}
        ~AsyncLopper()
        {
            stop();
        }
        void push(const char *Data, size_t len) // 将数据添加到缓冲区
        {
            // 1、无限扩容 --非安全  2、固定大小--缓冲区满了就阻塞
            unique_lock<mutex> lock(_mutex);
            // 若果缓冲区剩余空间大于要添加数据大小，则可以添加，否则进入等待队列
            if (_lopper_type == AsyncType::ASYNC_SAFE) // 如果不是安全状态，则不许要阻塞
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; });
            // 条件满足，可以向缓冲区添加数据
            _pro_buf.push(Data, len);
            // 唤醒消费线程
            _cond_con.notify_all();
        }
        void stop() // 关闭异步工作器
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒所有消费线程
            _thread.join();
        }

    private:
        Func Call_Back; // 数据处理回调函数
        // 线程入口函数 --对消费缓冲区中的数据进行处理
        void threadFunc()
        {
            while (1)
            {
                // 1、判断缓冲区是否存在数据
                {
                    unique_lock<mutex> lock(_mutex);
                    // 退出标志被设置，且缓冲区没有数据了，这时候在退出线程，避免生产缓冲区还有收据，但处理线程结束已经结束。
                    if (_stop && _pro_buf.empty())
                        break;
                    // 缓冲区没有数据，则消费线程进入等待队列等待资源就绪
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });

                    _con_buf.swap(_pro_buf);
                    if (_lopper_type == AsyncType::ASYNC_SAFE) // 如果不是安全状态，则不许要唤醒
                        // 2、唤醒生产者
                        _cond_pro.notify_all();
                }
                // 3、被唤醒后，对消费缓冲区数据进行处理
                Call_Back(_con_buf); //---->通过外部传入处理函数，进行数据处理
                // 4、初始化消费缓冲区
                _con_buf.reset();
            }
        }

    private:
        atomic<bool> _stop;           // 异步工作器状态标志
        AsyncType _lopper_type;       // 异步工作器处理模式
        LogBuffer::Buffer _pro_buf;   // 生产缓冲区
        LogBuffer::Buffer _con_buf;   // 消费缓冲区
        condition_variable _cond_pro; // 生产者条件变量
        condition_variable _cond_con; // 消费者条件变量 --生产缓冲区没有数据则消费者线程进入等待队列休眠
        mutex _mutex;
        thread _thread;
    };
}
#endif